package com.ttx.rabbitmq.demo.springboot;

import com.ttx.rabbitmq.demo.ResultHolder;
import com.ttx.rabbitmq.demo.config.RabbitMqConfig2;
import org.junit.Assert;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitOperations;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;

import java.util.concurrent.TimeUnit;

/**
 * 测试RabbitTemplate, 发送数据到rabbitmq
 * @author
 * @date
 */
@SpringBootTest
public class RabbitTemplateTests2 {
    @Autowired
    @Qualifier("defaultRabbitTemplate")
    RabbitTemplate defaultRabbitTemplate;


    // --------------------------- 测试confirm

    // 配置 spring.rabbitmq.publisher-confirm-type=correlated
    //测试confirm回调
    // 异步
    @Test
    public void sendWithConfirm() throws InterruptedException {
        //用于断言回调是否成功
        final ResultHolder resultHolder=new ResultHolder();

        RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                resultHolder.setResult(true);

                if(ack){
                    System.out.println(
                            String.format("confirm成功, ack:%s , cause:%s, correlationData:%s", ack, cause, correlationData)
                    );
                }else {
                    //nack
                    //可以放到一个阻塞队列, 另开线程, 重发
                }

            }
        };
        defaultRabbitTemplate.setConfirmCallback(confirmCallback);
        defaultRabbitTemplate.convertAndSend(RabbitMqConfig2.CONFIRM_QUEUE_NAME, "now 7777-1  confirm msg from defaultRabbitTemplate");

        System.out.println("等待3s");
        TimeUnit.SECONDS.sleep(3);

        Assert.assertTrue(resultHolder.getResult());
        System.out.println("结束");
    }



    // https://www.rabbitmq.com/tutorials/tutorial-seven-java.html
    // 配置 spring.rabbitmq.publisher-confirm-type=simple 或者 correlated
    // 同步
    @Test
    public void sendWithConfirm2() throws InterruptedException {
        RabbitOperations.OperationsCallback operationsCallback=(RabbitOperations operations)->{
            operations.convertAndSend(RabbitMqConfig2.CONFIRM_QUEUE_NAME, "now 7777-2 confirm msg from defaultRabbitTemplate");

            //在有限之间内等待确认，否则抛出异常
            operations.waitForConfirms(1000);
            return null;
        };
        defaultRabbitTemplate.invoke(operationsCallback);

        System.out.println("等待3s");
        TimeUnit.SECONDS.sleep(3);
        System.out.println("结束");
    }




    // --------------------------- 测试batch
    @Autowired
    BatchingRabbitTemplate batchQueueRabbitTemplate;

    @Test
    public void batchSend() throws InterruptedException {
        // 除了send(String exchange, String routingKey, Message message, CorrelationData correlationData)方法是发送单条数据
        // 其他send都是批量发送

        //批量发送


        long timestamp=System.currentTimeMillis();

        String msg;
        Message message;
        MessageProperties messageProperties=new MessageProperties();
        for(int i=0;i<1000;i++){
            msg="batch."+timestamp+"-"+i;


            message=new Message(msg.getBytes(), messageProperties);
            batchQueueRabbitTemplate.send(RabbitMqConfig2.BATCH_QUEUE_NAME,message);

//            defaultRabbitTemplate.convertAndSend(RabbitMqConfig2.BATCH_QUEUE_NAME, msg.getBytes());
        }

        System.out.println("发送数据完毕");
        System.out.println("等待30s");
        TimeUnit.SECONDS.sleep(30); //等待消费者消费
    }





    //-------------------------------- 测试 重试
    @Autowired
    RabbitTemplate retryRabbitTemplate;

    @Test
    public void retrySend(){
        //先关掉rabbitmq,然后在向rabbitmq发信息


        retryRabbitTemplate.convertAndSend(RabbitMqConfig2.RETRY_QUEUE_NAME,
                "retry-test.messages-"+System.currentTimeMillis());
    }


}
